package com.codeduck.handler;

import com.alibaba.fastjson.JSON;
import com.codeduck.pojo.SystemLog;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * @auther Mahone Yan
 * @date 2023/7/16
 */
@Component("SystemLogMsgHandler")
public class SystemLogMsgHandler extends AbstractMsgHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(SystemLogMsgHandler.class);

    /**
     * @param records
     */
    @Override
    public void batchConsume(ConsumerRecords<Object, Object> records) {
        int count = records.count();
        Iterator<ConsumerRecord<Object, Object>> iterator = records.iterator();
        List<Object> data = new ArrayList<>(count);
        while (iterator.hasNext()) {
            ConsumerRecord<Object, Object> record = iterator.next();
            data.add((record.value()));
        }
        LOGGER.info("系统日志已保存至数据库[{}], 数据来源于分区[{}]", JSON.toJSONString(data), JSON.toJSONString(records.partitions()));
    }
}
